package io.gitee.hfl.rocketmq.producer;

import com.alibaba.fastjson.JSON;
import io.gitee.hfl.rocketmq.annotation.RocketProducer;
import io.gitee.hfl.rocketmq.handler.ISendCallbackHandler;
import io.gitee.hfl.rocketmq.msg.RocketMessage;
import io.gitee.hfl.rocketmq.props.RocketProperties;
import io.gitee.hfl.rocketmq.resolver.PropertyResolver;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.SendReceipt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.io.IOException;
import java.util.concurrent.CompletableFuture;

/**
 * 通用的生产者基类
 *
 * @author hefulin
 * @date 2022/09/07 14:55
 */
public abstract class DefaultProducerProxy {

    @Resource
    private ClientConfiguration clientConfiguration;

    @Resource
    private ClientServiceProvider clientServiceProvider;

    @Resource
    private ISendCallbackHandler sendCallbackHandler;

    protected Producer producer;

    protected Logger log = LoggerFactory.getLogger(this.getClass());

    /**
     * 初始化 producer
     */
    @PostConstruct
    public void setup() throws ClientException {
        if (rocketProducer() == null || !rocketProducer().enable()) {
            log.warn("configuration is closed!");
            return;
        }
        producer = clientServiceProvider.newProducerBuilder()
                .setClientConfiguration(clientConfiguration)
                .build();
        log.info("create producer");
    }

    // 以下为同步发送

    /**
     * 同步发送
     *
     * @param msg   消息体
     * @param topic topic
     * @param tag   tag
     * @return 发送结果
     */
    public SendReceipt sendMsg(RocketMessage<?> msg, String topic, String tag) throws ClientException {
        this.printLog(msg);
        return producer.send(clientServiceProvider.newMessageBuilder()
                .setTopic(topic)
                .setTag(tag)
                .setBody(this.toByte(msg))
                .build());
    }

    /**
     * 同步发送
     *
     * @param msg   消息体
     * @param topic topic
     * @param tag   tag
     * @param key   业务主键
     * @return 发送结果
     */
    public SendReceipt sendMsg(RocketMessage<?> msg, String topic, String tag, String key) throws ClientException {
        this.printLog(msg);
        return producer.send(clientServiceProvider.newMessageBuilder()
                .setTopic(topic)
                .setTag(tag)
                .setKeys(key)
                .setBody(this.toByte(msg))
                .build());
    }

    /**
     * 同步发送
     *
     * @param message 消息体
     * @return 发送结果
     */
    public SendReceipt sendMsg(Message message) throws ClientException {
        return producer.send(message);
    }

    // 以下为异步发送

    /**
     * 异步发送消息
     *
     * @param msg   消息体
     * @param topic topic
     * @param tag   tag
     */
    public void sendMsgAsync(RocketMessage<?> msg, String topic, String tag) {
        this.printLog(msg);
        this.sendAsync(clientServiceProvider.newMessageBuilder()
                .setTopic(topic)
                .setTag(tag)
                .setBody(this.toByte(msg))
                .build());
    }

    /**
     * 异步发送消息
     *
     * @param msg   消息体
     * @param topic topic
     * @param tag   tag
     * @param key   业务主键
     */
    public void sendMsgAsync(RocketMessage<?> msg, String topic, String tag, String key) {
        this.printLog(msg);
        this.sendAsync(clientServiceProvider.newMessageBuilder()
                .setTopic(topic)
                .setTag(tag)
                .setKeys(key)
                .setBody(this.toByte(msg))
                .build());
    }

    // 以下为延时消息 单位: s

    /**
     * 异步发送延时消息
     *
     * @param msg     消息内容
     * @param topic   topic
     * @param tag     tag
     * @param timeout 延时时间
     */
    public void sendDelayMsgAsync(RocketMessage<?> msg, String topic, String tag, long timeout) {
        this.printLog(msg);
        this.sendAsync(clientServiceProvider.newMessageBuilder()
                .setTopic(topic)
                .setTag(tag)
                .setDeliveryTimestamp(System.currentTimeMillis() + timeout * 1000)
                .setBody(this.toByte(msg))
                .build());
    }

    /**
     * 异步发送延时消息
     *
     * @param msg     消息体
     * @param topic   topic
     * @param tag     tag
     * @param key     业务主键
     * @param timeout 延时时间
     */
    public void sendDelayMsgAsync(RocketMessage<?> msg, String topic, String tag, String key, long timeout) {
        printLog(msg);
        this.sendAsync(clientServiceProvider.newMessageBuilder()
                .setTopic(topic)
                .setTag(tag)
                .setDeliveryTimestamp(System.currentTimeMillis() + timeout * 1000)
                .setBody(this.toByte(msg))
                .setKeys(key)
                .build());
    }

    private void sendAsync(Message message) {
        CompletableFuture<SendReceipt> sendReceiptCompletableFuture = producer.sendAsync(message);
        sendReceiptCompletableFuture.whenCompleteAsync(sendCallbackHandler);
    }

    /**
     * json序列化成字节码
     *
     * @param o 消息内容
     * @return 字节数组
     */
    protected byte[] toByte(Object o) {
        return JSON.toJSONString(o).getBytes();
    }

    /**
     * 输出日志
     *
     * @param msg 消息内容
     * @param <T> 消息内容泛型
     */
    private <T> void printLog(T msg) {
        log.info("{} send rocket message:{} ", getClass().getSimpleName(), msg);
    }

    /**
     * 获取 producer 上面注册的ons配置注解
     *
     * @return 类上注解
     */
    protected RocketProducer rocketProducer() {
        return this.getClass().getAnnotation(RocketProducer.class);
    }

    @PreDestroy
    public void destroy() {
        try {
            producer.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        log.info("producer destroyed.");
    }

}

